package com.rn.gypsophila.rabbitmq.mq;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.concurrent.TimeUnit;

/**
 * Description: 抽象的分布式事务消息消费者
 *
 * @author rannuo
 * @date 2021/4/10
 */
@Slf4j
public abstract class AbstractTxMessageConsumer {

    @Autowired
    private Jackson2JsonMessageConverter messageConverter;

    /**
     * 最大消费次数
     */
    private static final Integer MAX_RETRY_COUNT = 3;

    /**
     * 本地缓存
     * TODO: 替换成分布式缓存，如Redis
     */
    private static final Cache<String, Integer> CACHE = Caffeine.newBuilder()
            .expireAfterAccess(3, TimeUnit.SECONDS)
            .maximumSize(100)
            .build();

    /**
     * 接收消息，子类必须实现该方法
     *
     * @param message          消息对象
     * @param messageConverter 消息转换器
     */
    public abstract void receiveMessage(Message message, MessageConverter messageConverter);

    /**
     * 处理消息
     */
    public void onMessage(Message message, Channel channel) throws Exception {
        final MessageProperties messageProperties = message.getMessageProperties();
        String messageId = (String) message.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
        long deliveryTag = messageProperties.getDeliveryTag();
        try {
            // 执行业务逻辑
            receiveMessage(message, messageConverter);
            // 手动签收消息
            channel.basicAck(deliveryTag, false);
            log.info("手动签收消息[{}],通知RabbitMQ服务器端删除该消息", messageId);
        } catch (Exception e) {
            log.error("消息[{}]消费处理异常，即将重试...", messageId, e);
            Integer retryCountInCache = CACHE.getIfPresent(messageId);
            int retryCount = retryCountInCache != null ? retryCountInCache : 0;
            if (retryCount >= MAX_RETRY_COUNT) {
                // 放入死信队列
                channel.basicReject(deliveryTag, false);
                log.info("消息[{}]经历{}次重试消费后仍未正常处理，放入死信队列", messageId, MAX_RETRY_COUNT);
            } else {
                // 放入消费队列（添加在队列的末端），重新消费
                channel.basicNack(deliveryTag, false, true);
                log.info("消息[{}]第{}次重试消费", messageId, retryCount);
                CACHE.put(messageId, retryCount + 1);
            }
        }
    }

}